package rocketmq

import (
	"context"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/star-find-cloud/star-chat/conf"
	"github.com/star-find-cloud/star-chat/pkg/database"
	log "github.com/star-find-cloud/star-chat/pkg/logger"
	"sync"
)

type EventConsumer struct {
	MQConsumer rocketmq.PushConsumer
	Topic      string
	cache      database.Redis
}

var (
	ConsumerMutex = &sync.Mutex{}
	clients       = map[string]rocketmq.PushConsumer{}
)

// NewConsumer 消费者：连接集群消费消息
func NewConsumer(topic string) (*EventConsumer, error) {
	ConsumerMutex.Lock()
	defer ConsumerMutex.Unlock()

	var config = conf.GetConfig()
	group := fmt.Sprintf("%s_%s_consumer_group", config.App.Name, topic)

	c, err := rocketmq.NewPushConsumer(
		consumer.WithNameServer([]string{config.MQ.RocketMQ.NameServer}),
		consumer.WithConsumerModel(consumer.Clustering), // 集群消费模式
		consumer.WithGroupName(group),
		consumer.WithMaxReconsumeTimes(3),
		consumer.WithRetry(3),
	)
	if err != nil {
		log.AppLogger.Errorf("create consumer error: %v\n", err)
		return nil, err
	}

	eventConsumer := &EventConsumer{
		MQConsumer: c,
		Topic:      topic,
	}
	if eventConsumer.MQConsumer == nil {
		log.MQLogger.Errorln("consumer  init failed, nil pointer")
		return nil, fmt.Errorf("consumer init failed, nil pointer")
	}
	clients[group] = c

	return eventConsumer, nil
}

// StartSubscribe 启动消费者, 并订阅主题
func (c *EventConsumer) StartSubscribe(topic, expression string, handler func(context.Context, ...*primitive.MessageExt) (consumer.ConsumeResult, error)) error {
	group := fmt.Sprintf("%s_%s_consumer_group", "app_name", topic)
	mqConsumer, exists := clients[group]
	if !exists {
		log.MQLogger.Errorf("consumer [%s group] not exists", topic)
	}

	err := mqConsumer.Subscribe(topic,
		consumer.MessageSelector{
			Type:       consumer.TAG,
			Expression: expression,
		},
		handler)
	if err != nil {
		log.MQLogger.Errorf("subscribe topic error: %v\n", err)
		return err
	}

	err = mqConsumer.Start()
	if err != nil {
		log.AppLogger.Errorf("start consumer error: %v\n", err)
		return err
	}
	log.MQLogger.Infof("consumer [%s group] start success", group)
	return nil
}

func (c *EventConsumer) Shutdown() error {
	err := c.MQConsumer.Shutdown()
	if err != nil {
		return err
	}
	return nil
}
